package com.chenqq.qsocket.mina;

import android.util.Log;

import com.chenqq.qsocket.adapter.TcpBasicAdapter;
import com.chenqq.qsocket.configuration.MinaConfigInfo;
import com.chenqq.qsocket.listener.SocketConnectListener;
import com.chenqq.qsocket.listener.SocketMessageListener;
import com.chenqq.qsocket.tools.BroadcastUnit;

import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.keepalive.KeepAliveFilter;
import org.apache.mina.filter.keepalive.KeepAliveMessageFactory;
import org.apache.mina.transport.socket.DatagramSessionConfig;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

import java.net.InetSocketAddress;

import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.BehaviorSubject;

/**
 * MINASocket服务端
 */
public class MinaSocketService {
    private NioSocketAcceptor nioSocketAcceptor;//监听器
    private InetSocketAddress inetSocketAddress;
    private String tag = MinaSocketService.class.getSimpleName();
    private Disposable subscribeTcpService;
    private SocketConnectListener connectListener;
    private SocketMessageListener socketMessageListener;

    public MinaSocketService() {
        this.nioSocketAcceptor = new NioSocketAcceptor();
    }

    public MinaSocketService setHandler(IoHandlerAdapter ioHandlerAdapter) {
        nioSocketAcceptor.setHandler(ioHandlerAdapter);
        return this;
    }

    public MinaSocketService setFilterChain(String name, IoFilter filter) {
        nioSocketAcceptor.getFilterChain().addLast(name, filter);
        return this;
    }


    public MinaSocketService setReadBufferSize(int size) {
        nioSocketAcceptor.getSessionConfig().setReadBufferSize(size);
        return this;
    }


    public MinaSocketService setIdleTime(IdleStatus idleStatus, int time) {
        nioSocketAcceptor.getSessionConfig().setIdleTime(idleStatus, time);
        return this;
    }

    public MinaSocketService setSocketMessageListener(SocketMessageListener socketMessageListener) {
        this.socketMessageListener = socketMessageListener;
        return this;
    }

    public MinaSocketService setKeepAlive(boolean isKeepAlive) {
        nioSocketAcceptor.getSessionConfig().setKeepAlive(isKeepAlive);
        return this;
    }

    public MinaSocketService setKeepAliveFilter(int intervalTime, int timeOut
            , KeepAliveMessageFactory keepAliveMessageFactory) {
        KeepAliveFilter heartFilter = new KeepAliveFilter(keepAliveMessageFactory, IdleStatus.BOTH_IDLE);
        //每 5 s发送一个心跳包
        heartFilter.setRequestInterval(intervalTime);
        heartFilter.setForwardEvent(false);
        //心跳包超时时间 10s
        heartFilter.setRequestTimeout(timeOut);
        nioSocketAcceptor.getFilterChain().addLast("heartbeat", heartFilter);
        return this;
    }


    public MinaSocketService setTcpNoDelay(boolean isDelay){
        nioSocketAcceptor.getSessionConfig().setTcpNoDelay(isDelay);
        return this;
    }

    public MinaSocketService setMinaPort(int port) {
        inetSocketAddress = new InetSocketAddress(port);
        return this;
    }

    public MinaSocketService setReuseAddress(boolean isReuse){
        nioSocketAcceptor.setReuseAddress(true);
        return this;
    }
    public MinaSocketService setConnectListener(SocketConnectListener connectListener) {
        this.connectListener = connectListener;
        return this;
    }

    public MinaSocketService setTag(String tag) {
        this.tag = tag;
        return this;
    }

    public void start() {
        if (subscribeTcpService == null) {
            subscribeTcpService = BehaviorSubject.create(new ObservableOnSubscribe<Boolean>() {
                @Override
                public void subscribe(@NonNull ObservableEmitter<Boolean> emitter) throws Exception {
                    emitter.onNext(true);
                }
            }).unsubscribeOn(Schedulers.io())
                    .subscribeOn(Schedulers.io())
                    .map(new Function<Boolean, String>() {
                        @Override
                        public String apply(@NonNull Boolean o) throws Exception {
                            if (nioSocketAcceptor == null) {
                                return "没有初始化";
                            }
                            if (inetSocketAddress == null) {
                                return "没有初始化端口";
                            }
                            IoHandler handler = nioSocketAcceptor.getHandler();
                            if (handler instanceof TcpBasicAdapter){
                                ((TcpBasicAdapter) handler).setSocketMessageListener(socketMessageListener)
                                        .setTag(tag).setFileDown(MinaConfigInfo.getInStance().getTemporaryPath()
                                ,false)
                                        .setFileDown(MinaConfigInfo.getInStance().getPathDown(),true);
                            }
                            nioSocketAcceptor.bind(inetSocketAddress);
                            String broadcast = BroadcastUnit.ipAddress();
                            return "Success:"+ broadcast+ " 端口："+inetSocketAddress.getPort();
                        }
                    }).observeOn(Schedulers.io())
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(String s) throws Exception {
                            Log.i("" + tag, s);
                            if (s.startsWith("Success")) {
                                if (connectListener != null) {
                                    connectListener.isConnect();
                                }
                            } else {
                                if (connectListener != null) {
                                    connectListener.connectFailed(tag, s);
                                }
                            }

                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            if (connectListener != null) {
                                connectListener.connectFailed(tag, throwable.toString());
                            }
                        }
                    });
        }

    }

    public void stop() {
        if (nioSocketAcceptor != null) {
            nioSocketAcceptor.unbind();
            nioSocketAcceptor.dispose();
            nioSocketAcceptor = null;
        }
        if (subscribeTcpService != null) {
            subscribeTcpService.dispose();
            subscribeTcpService = null;
        }
    }


}
